/*
 * Copyright 2002-2019 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.summerframework.component.quartz.annotation;

import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.quartz.impl.JobDetailImpl;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.*;
import org.springframework.beans.support.ArgumentConvertingMethodInvoker;
import org.springframework.cglib.core.SpringNamingPolicy;
import org.springframework.cglib.proxy.*;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.MethodInvoker;

import java.lang.reflect.InvocationTargetException;

/**
 * {@link FactoryBean} that exposes a
 * {@link JobDetail} object which delegates job execution to a
 * specified (static or non-static) method. Avoids the need for implementing
 * a one-line Quartz Job that just invokes an existing service method on a
 * Spring-managed target bean.
 *
 * <p>Inherits common configuration properties from the {@link MethodInvoker}
 * base class, such as {@link #setTargetObject "targetObject"} and
 * {@link #setTargetMethod "targetMethod"}, adding support for lookup of the target
 * bean by name through the {@link #setTargetBeanName "targetBeanName"} property
 * (as alternative to specifying a "targetObject" directly, allowing for
 * non-singleton target objects).
 *
 * <p>Supports both concurrently running jobs and non-currently running
 * jobs through the "concurrent" property. Jobs created by this
 * MethodInvokingJobDetailFactoryBean are by default volatile and durable
 * (according to Quartz terminology).
 *
 * <p><b>NOTE: JobDetails created via this FactoryBean are <i>not</i>
 * serializable and thus not suitable for persistent job stores.</b>
 * You need to implement your own Quartz Job as a thin wrapper for each case
 * where you want a persistent job to delegate to a specific service method.
 *
 * <p>Compatible with Quartz 2.1.4 and higher, as of Spring 4.1.
 *
 * @author Juergen Hoeller
 * @author Alef Arendsen
 * @see #setTargetBeanName
 * @see #setTargetObject
 * @see #setTargetMethod
 * @see #setConcurrent
 * @since 18.02.2004
 */
@Slf4j
public class BeanMethodInvokingJobDetailFactoryBean extends ArgumentConvertingMethodInvoker
        implements FactoryBean<JobDetail>, BeanNameAware, BeanClassLoaderAware, BeanFactoryAware, InitializingBean {

    public static ThreadLocal<JobExecutionContext> jobExecutionContextThreadLocal = new ThreadLocal<>();

    @Nullable
    private String name;

    private String group = Scheduler.DEFAULT_GROUP;

    private boolean concurrent = true;

    @Nullable
    private String targetBeanName;

    @Nullable
    private String beanName;

    @Nullable
    private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();

    @Nullable
    private BeanFactory beanFactory;

    @Nullable
    private JobDetail jobDetail;

    public static JobExecutionContext getJobExecutionContext() {
        return jobExecutionContextThreadLocal.get();
    }

    /**
     * Set the name of the job.
     * <p>Default is the bean name of this FactoryBean.
     */
    public void setName(String name) {
        this.name = name;
    }

    /**
     * Set the group of the job.
     * <p>Default is the default group of the Scheduler.
     *
     * @see Scheduler#DEFAULT_GROUP
     */
    public void setGroup(String group) {
        this.group = group;
    }

    /**
     * Specify whether or not multiple jobs should be run in a concurrent fashion.
     * The behavior when one does not want concurrent jobs to be executed is
     * realized through adding the {@code @PersistJobDataAfterExecution} and
     * {@code @DisallowConcurrentExecution} markers.
     * More information on stateful versus stateless jobs can be found
     * <a href="https://www.quartz-scheduler.org/documentation/quartz-2.3.0/tutorials/tutorial-lesson-03.html">here</a>.
     * <p>The default setting is to run jobs concurrently.
     */
    public void setConcurrent(boolean concurrent) {
        this.concurrent = concurrent;
    }

    /**
     * Set the name of the target bean in the Spring BeanFactory.
     * <p>This is an alternative to specifying {@link #setTargetObject "targetObject"},
     * allowing for non-singleton beans to be invoked. Note that specified
     * "targetObject" and {@link #setTargetClass "targetClass"} values will
     * override the corresponding effect of this "targetBeanName" setting
     * (i.e. statically pre-define the bean type or even the bean object).
     */
    public void setTargetBeanName(String targetBeanName) {
        this.targetBeanName = targetBeanName;
    }

    @Override
    public void setBeanName(String beanName) {
        this.beanName = beanName;
    }

    @Override
    public void setBeanClassLoader(ClassLoader classLoader) {
        this.beanClassLoader = classLoader;
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }

    @Override
    protected Class<?> resolveClassName(String className) throws ClassNotFoundException {
        return ClassUtils.forName(className, this.beanClassLoader);
    }


    @Override
    public void afterPropertiesSet() throws ClassNotFoundException, NoSuchMethodException {
        prepare();

        // Use specific name if given, else fall back to bean name.
        String name = (this.name != null ? this.name : this.beanName);

        // Consider the concurrent flag to choose between stateful and stateless job.
        Class<? extends Job> jobClass = (this.concurrent ? MethodInvokingJob.class : StatefulMethodInvokingJob.class);

        Enhancer enhancer = new Enhancer();
        enhancer.setSuperclass(jobClass);
        enhancer.setNamingPolicy(new SpringNamingPolicy() {
            @Override
            protected String getTag() {
                return name;
            }
        });
        enhancer.setCallbackType(NoOp.class);
        enhancer.setUseCache(false);
//        enhancer.setCallback((MethodInterceptor) (obj, method, args, proxy) -> proxy.invokeSuper(obj, args));
        final Class aClass = enhancer.createClass();

        // Build JobDetail instance.
        JobDetailImpl jdi = new JobDetailImpl();
        jdi.setName(name != null ? name : toString());
        jdi.setGroup(this.group);
        jdi.setJobClass(aClass);
        jdi.setDurability(true);
        jdi.getJobDataMap().put("targetClass", getTargetClass().getName());
        jdi.getJobDataMap().put("targetBean", targetBeanName);
        jdi.getJobDataMap().put("targetMethod", getTargetMethod());
        this.jobDetail = jdi;

        postProcessJobDetail(this.jobDetail);
    }

    /**
     * Callback for post-processing the JobDetail to be exposed by this FactoryBean.
     * <p>The default implementation is empty. Can be overridden in subclasses.
     *
     * @param jobDetail the JobDetail prepared by this FactoryBean
     */
    protected void postProcessJobDetail(JobDetail jobDetail) {
    }


    /**
     * Overridden to support the {@link #setTargetBeanName "targetBeanName"} feature.
     */
    @Override
    public Class<?> getTargetClass() {
        Class<?> targetClass = super.getTargetClass();
        if (targetClass == null && this.targetBeanName != null) {
            Assert.state(this.beanFactory != null, "BeanFactory must be set when using 'targetBeanName'");
            targetClass = this.beanFactory.getType(this.targetBeanName);
        }
        return targetClass;
    }

    /**
     * Overridden to support the {@link #setTargetBeanName "targetBeanName"} feature.
     */
    @Override
    public Object getTargetObject() {
        Object targetObject = super.getTargetObject();
        if (targetObject == null && this.targetBeanName != null) {
            Assert.state(this.beanFactory != null, "BeanFactory must be set when using 'targetBeanName'");
            targetObject = this.beanFactory.getBean(this.targetBeanName);
        }
        return targetObject;
    }


    @Override
    @Nullable
    public JobDetail getObject() {
        return this.jobDetail;
    }

    @Override
    public Class<? extends JobDetail> getObjectType() {
        return (this.jobDetail != null ? this.jobDetail.getClass() : JobDetail.class);
    }

    @Override
    public boolean isSingleton() {
        return true;
    }


    /**
     * Quartz Job implementation that invokes a specified method.
     * Automatically applied by MethodInvokingJobDetailFactoryBean.
     */
    public static class MethodInvokingJob extends QuartzJobBean implements ApplicationContextAware {

        private String targetBean;

        private String targetMethod;

        private String targetClass;

        private ApplicationContext applicationContext;

        public void setTargetBean(String targetBean) {
            this.targetBean = targetBean;
        }

        public void setTargetMethod(String targetMethod) {
            this.targetMethod = targetMethod;
        }

        public void setTargetClass(String targetClass) {
            this.targetClass = targetClass;
        }

        /**
         * Invoke the method via the MethodInvoker.
         */
        @Override
        protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
            Assert.state(this.targetBean != null && this.targetMethod != null, "No MethodInvoker set");
            jobExecutionContextThreadLocal.set(context);
            try {
                Object bean = applicationContext.getBean(targetBean, Class.forName(targetClass));
                MethodInvoker methodInvoker = new MethodInvoker();
                methodInvoker.setTargetObject(bean);
                methodInvoker.setTargetMethod(targetMethod);
                methodInvoker.prepare();

                context.setResult(methodInvoker.invoke());
            } catch (InvocationTargetException ex) {
                if (ex.getTargetException() instanceof JobExecutionException) {
                    throw (JobExecutionException) ex.getTargetException();
                } else {
                    final JobExecutionException jobExecutionException = new JobExecutionException("Invocation of method '" + targetMethod + "' on target class [" + targetClass + "] failed", ex);
                    jobExecutionException.setRefireImmediately(true);

                    throw new RuntimeException("Invocation of method '" + targetMethod + "' on target class [" + targetClass + "] failed", ex);
                }
            } catch (Exception ex) {
                final JobExecutionException jobExecutionException = new JobExecutionException("Invocation of method '" + targetMethod + "' on target class [" + targetClass + "] failed", ex);
                jobExecutionException.setRefireImmediately(true);

                throw jobExecutionException;
            } finally {
                jobExecutionContextThreadLocal.remove();
            }
        }

        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
        }
    }

    /**
     * Extension of the MethodInvokingJob, implementing the StatefulJob interface.
     * Quartz checks whether or not jobs are stateful and if so,
     * won't let jobs interfere with each other.
     */
    @PersistJobDataAfterExecution
    @DisallowConcurrentExecution
    public static class StatefulMethodInvokingJob extends MethodInvokingJob {

        // No implementation, just an addition of the tag interface StatefulJob
        // in order to allow stateful method invoking jobs.
    }

}
